package com.vhall.component.service.room.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.vhall.component.dao.room.RoomJoinsMapper;
import com.vhall.component.entity.common.CachePage;
import com.vhall.component.entity.rebroadcast.RebroadCastEntity;
import com.vhall.component.entity.record.RecordEntity;
import com.vhall.component.entity.room.dto.RoomJoinsCacheDTO;
import com.vhall.component.entity.room.dto.RoomsEntityDTO;
import com.vhall.component.entity.room.entity.RoomExtendsEntity;
import com.vhall.component.entity.room.entity.RoomJoinsEntity;
import com.vhall.component.entity.room.entity.RoomsEntity;
import com.vhall.component.entity.room.enums.RoomsWarmTypeEnum;
import com.vhall.component.entity.room.vo.BannedKickedRspVO;
import com.vhall.component.framework.common.exception.BusinessException;
import com.vhall.component.framework.common.utils.CollectionUtils;
import com.vhall.component.framework.common.utils.StringUtils;
import com.vhall.component.plugin.common.constant.RedisKey;
import com.vhall.component.plugin.common.constant.RoomConstant;
import com.vhall.component.plugin.common.exception.BizErrorCode;
import com.vhall.component.plugin.common.utils.JsonUtil;
import com.vhall.component.service.room.RoomCacheLockService;
import com.vhall.component.service.room.RoomCacheService;
import com.vhall.component.service.room.assembler.RoomJoinsAssembler;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author yuanzh
 */

@Slf4j
@Service
public class RoomCacheServiceImpl implements RoomCacheService {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RoomJoinsAssembler roomJoinsAssembler;

    @Autowired
    private RoomJoinsMapper roomJoinsMapper;

    @Autowired
    private RoomCacheLockService roomCacheLockService;

    /**
     * 非高频率  主要用于一些更新
     */
    @Override
    public void cacheRoomJoins(Integer joinId) {
        if (Objects.nonNull(joinId)) {
            cacheRoomJoins(Collections.singletonList(joinId));
        }
    }

    @Override
    public void cacheRoomJoins(String roomId, Integer accountId) {
        cacheRoomJoins(roomId, Collections.singletonList(accountId));
    }

    @Override
    public void cacheRoomJoins(RoomJoinsEntity roomJoins) {
        if (Objects.nonNull(roomJoins)) {
            //缓存参会信息
            handleCacheRoomJoins(Collections.singletonList(roomJoins));
        }
    }

    @Override
    public RoomJoinsEntity getCachedRoomJoins(String roomId, Integer accountId) {

        //查询 redis缓存  如果没有查到 查询数据库
        Object joinInfo = redisTemplate.opsForValue().get(RedisKey.ROOM_JOIN_DATA_CACHE_PRE_KEY + roomId + accountId);
        if (Objects.nonNull(joinInfo)) {
            return JsonUtil.objectFromJSONString(joinInfo.toString(), RoomJoinsEntity.class);
        }


        LambdaQueryWrapper<RoomJoinsEntity> wrapper = Wrappers.lambdaQuery();
        //代表强制到主库
        wrapper.first("/*FORCE_MASTER*/");
        wrapper.eq(RoomJoinsEntity::getRoomId, roomId);
        wrapper.eq(RoomJoinsEntity::getAccountId, accountId);
        RoomJoinsEntity roomJoins = roomJoinsMapper.selectOne(wrapper);

        if (Objects.nonNull(roomJoins)) {
            handleCacheRoomJoins(Collections.singletonList(roomJoins));
        } else {
            return null;
        }
        return roomJoins;

    }


    @Override
    public void clearHostRoomJoin(String roomId) {
        redisTemplate.delete(RedisKey.ROOM_JOIN_INFO_CACHE + roomId);
    }

    @Override
    public void cacheHostRoomJoin(RoomJoinsEntity entity) {
        if (Objects.nonNull(entity) && StringUtils.isNotEmpty(entity.getRoomId())) {
            redisTemplate.opsForValue().set(RedisKey.ROOM_JOIN_INFO_CACHE + entity.getRoomId(), JsonUtil.toJsonString(entity));
        }
    }

    @Override
    public RoomJoinsEntity getCachedHostRoomJoin(String roomId) {
        Object joinInfo = redisTemplate.opsForValue().get(RedisKey.ROOM_JOIN_INFO_CACHE + roomId);
        if (Objects.nonNull(joinInfo)) {
            return JsonUtil.objectFromJSONString(joinInfo.toString(), RoomJoinsEntity.class);
        }
        return null;
    }


    /**
     * 根据joinId缓存房间参会信息
     *
     * @param joinIds 参会Ids
     * @return
     */
    @Override
    public int cacheRoomJoins(List<Integer> joinIds) {

        //这个主要用于通过  joinId 查询主库 把 数据查询出来，然后 缓存 即可
        LambdaQueryWrapper<RoomJoinsEntity> queryWrapper = Wrappers.lambdaQuery();
        //查询主库
        queryWrapper.first("/*FORCE_MASTER*/");
        queryWrapper.in(RoomJoinsEntity::getJoinId, joinIds);

        List<RoomJoinsEntity> entities = roomJoinsMapper.selectList(queryWrapper);

        if (CollectionUtils.isEmpty(entities)) {
            log.error("缓存room_joins信息失败，数据为空，无法处理");
            throw new BusinessException(BizErrorCode.EMPTY_DATA);
        }
        //缓存数据
        return handleCacheRoomJoins(entities);
    }

    /**
     * 非高频
     * 使用场景：禁言、踢人会调用到
     * 根据房间id与账户ids缓存RoomJoins
     *
     * @param roomId     房间id
     * @param accountIds 用户ids
     * @return
     */
    @Override
    public int cacheRoomJoins(String roomId, List<Integer> accountIds) {
        LambdaQueryWrapper<RoomJoinsEntity> queryWrapper = Wrappers.lambdaQuery();
        //查询主库
        queryWrapper.first("/*FORCE_MASTER*/");
        queryWrapper.eq(RoomJoinsEntity::getRoomId, roomId).in(RoomJoinsEntity::getAccountId, accountIds);
        List<RoomJoinsEntity> entities = roomJoinsMapper.selectList(queryWrapper);
        if (CollectionUtils.isEmpty(entities)) {
            log.error("缓存room_joins信息失败，数据为空，无法处理");
            throw new BusinessException(BizErrorCode.EMPTY_DATA);
        }
        //查询之后在缓存
        return handleCacheRoomJoins(entities);
    }


    /**
     * 新改造的缓存房间参会信息的方法
     *
     * @param entities 房间参会信息
     * @return
     */
    private int handleCacheRoomJoins(List<RoomJoinsEntity> entities) {
        List<Integer> joinIds = new ArrayList<>();
        int count = 0;
        String cacheKey;
        for (RoomJoinsEntity entity : entities) {
            String roomId = entity.getRoomId();

            cacheKey = RedisKey.ROOM_JOIN_DATA_CACHE_PRE_KEY + roomId + entity.getAccountId();
            RoomJoinsCacheDTO joinsCache = roomJoinsAssembler.joinEntityToCacheDTO(entity);

            //缓存 数据3天即可
            redisTemplate.opsForValue().set(cacheKey, JsonUtil.toJsonString(joinsCache), 3, TimeUnit.DAYS);


            joinIds.add(entity.getJoinId());
            // 若为主持人参会，则单独缓存主持人参会信息
            if (entity.getRoleName().equals(RoomConstant.HOST)) {
                cacheHostRoomJoin(entity);
            }
            count++;
        }
        log.info("缓存room_joins信息完成，joinIds={}，count={}", joinIds, count);
        return count;
    }

    @Override
    public void cacheKickedJoin(String roomId, List<Integer> accountIds, boolean isKicked) {
        handleCacheKickedOrBanned(roomId, accountIds, isKicked, 0);
        cacheRoomJoins(roomId, accountIds);
    }

    @Override
    public void cacheBannedJoin(String roomId, List<Integer> accountIds, boolean isBanned) {
        handleCacheKickedOrBanned(roomId, accountIds, isBanned, 1);
        cacheRoomJoins(roomId, accountIds);
    }

    @Override
    public CachePage<BannedKickedRspVO> listKickedCache(String roomId, Integer page, Integer pageSize) {
        return handleListKickedBannedCache(roomId, page, pageSize, 0);
    }

    @Override
    public CachePage<BannedKickedRspVO> listBannedCache(String roomId, Integer page, Integer pageSize) {
        return handleListKickedBannedCache(roomId, page, pageSize, 1);
    }

    /**
     * 公共处理禁言或踢出缓存
     *
     * @param roomId     房间Id
     * @param accountIds 用户ids
     * @param flag       标记，true=踢出&禁言 / false=取消踢出&取消禁言
     * @param opType     操作类型，0=踢出类 1=禁言类
     */
    private void handleCacheKickedOrBanned(String roomId, List<Integer> accountIds, boolean flag, int opType) {
        if (opType != 0 && opType != 1) {
            log.error("处理禁言/踢出失败，操作不支持");
            throw new BusinessException(BizErrorCode.BIZ_TYPE_NOT_ALLOWED);
        }

        String sortedCacheKey = RedisKey.ROOM_JOIN_KICKED_IDS_CACHE_PRE_KEY + roomId;
        if (opType == 1) {
            sortedCacheKey = RedisKey.ROOM_JOIN_BANNED_IDS_CACHE_PRE_KEY + roomId;
        }

        List<String> accountIdStringList = accountIds.stream().map(String::valueOf).collect(Collectors.toList());

        // 遍历参会信息，放入或清理
        BoundZSetOperations boundOps = redisTemplate.boundZSetOps(sortedCacheKey);
        if (flag) {
            HashSet<ZSetOperations.TypedTuple<String>> typedTuples = new HashSet<>();
            for (String accountId : accountIdStringList) {
                typedTuples.add(new DefaultTypedTuple<>(accountId, (double) System.currentTimeMillis()));
            }
            boundOps.add(typedTuples);
        } else {
            // 取消操作
            Long removeCount = boundOps.remove(accountIdStringList.toArray());
            log.info("公共处理禁言或踢出缓存，roomId={}，accountIds={}，removeCount={}", roomId, accountIdStringList, removeCount);
        }
    }

    /**
     * 公共处理获取禁言或踢出列表
     *
     * @param roomId   房间id
     * @param page     页码
     * @param pageSize 条数
     * @param opType   操作类型，0=踢出类 1=禁言类
     * @return 数据
     */
    public CachePage<BannedKickedRspVO> handleListKickedBannedCache(String roomId, Integer page, Integer pageSize, int opType) {
        if (opType != 0 && opType != 1) {
            log.error("获取禁言/踢出列表失败，操作不支持");
            throw new BusinessException(BizErrorCode.BIZ_TYPE_NOT_ALLOWED);
        }
        String idsCacheKey = RedisKey.ROOM_JOIN_KICKED_IDS_CACHE_PRE_KEY + roomId;
        if (opType == 1) {
            idsCacheKey = RedisKey.ROOM_JOIN_BANNED_IDS_CACHE_PRE_KEY + roomId;
        }

        CachePage<BannedKickedRspVO> cachePage = new CachePage<>();
        cachePage.setTotal(0);
        cachePage.setCurrent(page);
        cachePage.setPageSize(pageSize);
        cachePage.setList(Collections.emptyList());
        // 计算总条数
        BoundZSetOperations boundIdsOps = redisTemplate.boundZSetOps(idsCacheKey);
        Long count = boundIdsOps.zCard();
        if (Objects.isNull(count)) {
            return cachePage;
        }
        cachePage.setTotal(count);


        //  这块 代码 需要改
        //   这个是一堆账号id
        Set set = boundIdsOps.reverseRange((long) (page - 1) * pageSize, page * pageSize - 1L);
        if (CollectionUtils.isNotEmpty(set)) {
            List<String> accountList = new ArrayList<String>(set);

            LambdaQueryWrapper<RoomJoinsEntity> queryWrapper = Wrappers.lambdaQuery();
            //查询主库
            queryWrapper.first("/*FORCE_MASTER*/");
            queryWrapper.eq(RoomJoinsEntity::getRoomId, roomId);
            queryWrapper.in(RoomJoinsEntity::getAccountId, accountList);

            //  查询 所有的参会信息  带 账号id的
            List<RoomJoinsEntity> roomJoinsEntities = roomJoinsMapper.selectList(queryWrapper);


            if (roomJoinsEntities != null) {
                if (roomJoinsEntities.size() > 0) {
                    List<BannedKickedRspVO> resp = roomJoinsAssembler.joinsEntitiesToRspVos(roomJoinsEntities);
                    cachePage.setList(resp);
                }
            }

        }
        return cachePage;
    }

    @Override
    public Set<Integer> listCachedWaitingForWheatAccountId(String roomId) {
        String cacheKey = RedisKey.INTERACT_WAITING_FOR_WHEAT + roomId;
        Set<String> accountIds = stringRedisTemplate.opsForZSet().range(cacheKey, 0, -1);
        if (Objects.isNull(accountIds) || accountIds.isEmpty()) {
            return Collections.emptySet();
        }
        return accountIds.stream().map(Integer::parseInt).collect(Collectors.toSet());
    }

    @Override
    public Map<String, Object> getCachedRoomWarmInfo(RoomsEntityDTO room) {
        if (Objects.isNull(room)) {
            throw new BusinessException(BizErrorCode.EMPTY_ROOM);
        }
        Integer ilId = room.getIlId();
        Map<String, Object> warmInfo = new HashMap<>(8);
        // 默认暖场为图片
        warmInfo.put("type", 0);
        // 处理暖场信息
        Object cachedWarm = redisTemplate.opsForValue().get(RedisKey.WARM_INFO + ilId);
        if (Objects.nonNull(cachedWarm)) {
            Map<String, Object> cachedWarmInfoMap = JsonUtil.string2Map(cachedWarm.toString());
            if (CollectionUtils.isNotEmpty(cachedWarmInfoMap)) {
                warmInfo.putAll(cachedWarmInfoMap);
            }
            String typeKey = "type";
            // 若缓存的暖场类型为图片，直接返回
            if (warmInfo.containsKey(typeKey) && "0".equals(String.valueOf(warmInfo.get(typeKey)))) {
                return warmInfo;
            }
            String warmVodId = room.getWarmVodId();
            // 若当前房间暖场为视频，继续判断音视频存在与否
            if (room.getWarmType() == 1 && StringUtils.isNotEmpty(warmVodId)) {
                Integer warmType = room.getWarmType();
                JSONObject cacheObj = buildWarmInfoCache(ilId, warmVodId, warmType);
                if (cacheObj != null) return cacheObj;
            }
        }else{
            String warmVodId = room.getWarmVodId();
            // 若当前房间暖场为视频，继续判断音视频存在与否
            Integer warmType = room.getWarmType();
            JSONObject cacheObj = buildWarmInfoCache(ilId, warmVodId, warmType);
            if (cacheObj != null) return cacheObj;
        }
        return warmInfo;
    }

    @Nullable
    private JSONObject buildWarmInfoCache(Integer ilId, String warmVodId, Integer warmType) {
        if (Objects.equals(warmType,RoomsWarmTypeEnum.VIDEO.getValue()) && StringUtils.isNotEmpty(warmVodId)) {
            RecordEntity recordEntity = roomCacheLockService.getCachedWarmRecordInfo(warmVodId);
            // 若对应音视频不存在，重新缓存，并处理返回暖场视频id为空
            if (Objects.isNull(recordEntity)) {
                JSONObject cacheObj = new JSONObject();
                cacheObj.put("type", RoomsWarmTypeEnum.IMAGE.getValue());
                cacheObj.put("vod_name", "");
                cacheObj.put("vod_id", "");
                cacheObj.put("img", "");
                redisTemplate.opsForValue().set(RedisKey.WARM_INFO + ilId, cacheObj.toJSONString());
                return cacheObj;
            }else{
                JSONObject cacheObj = new JSONObject();
                cacheObj.put("type", RoomsWarmTypeEnum.VIDEO.getValue());
                cacheObj.put("vod_name", recordEntity.getName());
                cacheObj.put("vod_id", recordEntity.getVodId());
                cacheObj.put("img", "");
                redisTemplate.opsForValue().set(RedisKey.WARM_INFO + ilId, cacheObj.toJSONString());
                return cacheObj;
            }
        }
        return null;
    }


    @Override
    public RebroadCastEntity getRebroadCache(String roomId) {
        Object rebroadcast = redisTemplate.opsForValue().get(RedisKey.REBROADCAST_INFO + roomId);
        if (Objects.nonNull(rebroadcast)) {
            return JsonUtil.objectFromJSONString(rebroadcast.toString(), RebroadCastEntity.class);
        }
        return null;
    }

    @Override
    public void cacheRebroad(RebroadCastEntity rebroadCastEntity) {
        if (Objects.nonNull(rebroadCastEntity) && StringUtils.isNotEmpty(rebroadCastEntity.getRoomId())) {
            redisTemplate.opsForValue().set(RedisKey.REBROADCAST_INFO + rebroadCastEntity.getRoomId(), JsonUtil.toJsonString(rebroadCastEntity));
        }
    }

    @Override
    public RoomExtendsEntity getRoomExtendCache(String roomId) {
        Object roomExtend = redisTemplate.opsForValue().get(RedisKey.ROOM_EXTEND_CACHE + roomId);
        if (Objects.nonNull(roomExtend)) {
            return JsonUtil.objectFromJSONString(roomExtend.toString(), RoomExtendsEntity.class);
        }
        return null;
    }

    @Override
    public void cacheRoomExtend(RoomExtendsEntity roomExtendsEntity) {
        if (Objects.nonNull(roomExtendsEntity) && StringUtils.isNotEmpty(roomExtendsEntity.getRoomId())) {
            redisTemplate.opsForValue().set(RedisKey.ROOM_EXTEND_CACHE + roomExtendsEntity.getRoomId(), JsonUtil.toJsonString(roomExtendsEntity));
        }
    }
}
